package org.tiger.springboot.rocketMq.producer;

import javax.annotation.PostConstruct;

import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

/**
 * 消息生产者
 * 
 * @author liuwenlai
 *
 */
@Component
public class RocketMqProducer {

	private static final Logger logger = LoggerFactory.getLogger(RocketMqProducer.class);

	@Value("${spring.rocketmq.namesrvaddr}")
	private String namesrvAddr;

	@Value("${spring.rocketmq.timeout:60000}")
	private long sendMessageTimeOut;

	// 生产者组
	private final DefaultMQProducer defaultMQProducer = new DefaultMQProducer("TestRocketMQProducer");

	/**
	 * 初始化生产者
	 */
	@PostConstruct
	public void start() {
		try {
			logger.info("########## MQ produce start");
			defaultMQProducer.setNamesrvAddr(namesrvAddr);
			defaultMQProducer.start();
		} catch (Exception e) {
			logger.error("########## MQ produce start error:", e);
			throw new RuntimeException(e.getMessage(), e);
		}
	}

	/**
	 * 发送消息
	 * 
	 * @param data
	 * @param topic
	 * @param tags
	 * @param keys
	 */
	public void sendMessage(String data, String topic, String tags, String keys) {
		try {
			byte[] messageBody = data.getBytes(RemotingHelper.DEFAULT_CHARSET);

			Message message = new Message(topic, tags, keys, messageBody);

			defaultMQProducer.send(message, new SendCallback() {

				@Override
				public void onSuccess(SendResult paramSendResult) {
					logger.info("send message success,messageId:", paramSendResult.getMsgId());

				}

				@Override
				public void onException(Throwable paramThrowable) {
					logger.error("send message error:", paramThrowable.getMessage());

				}
			}, sendMessageTimeOut);

		} catch (Exception e) {
			logger.error("send message error:", e);
		}
	}

}
